package com.mininglamp.kafkastream;

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.TopologyBuilder;

import java.util.Properties;

/**
 * Project: ECommerceRecommendSystem
 * Package: com.mininglamp.kafkastream
 * Description
 * <p>
 * Created by ZhouPeng on 2022/01/07 17:54
 **/
public class Application {
    public static void main(String[] args) {

        String brokers = "master:9092,slave1:9092,slave2:9092";
        String zookeepers = "master:2181,slave1:2181,slave2:2181";

        //定义输入输出topic
        String from = "log";
        String to = "recommender";

        //定义kafka stream 配置参数
        Properties properties = new Properties();
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"logFilter");
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,brokers);
        properties.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG,zookeepers);

        //创建kafka stream配置对象
        StreamsConfig streamsConfig = new StreamsConfig(properties);

        //定义拓扑构建器
        TopologyBuilder builder = new TopologyBuilder();
        builder.addSource("SOURCE",from)
                .addProcessor("PROCESSOR",() -> new LogProcessor(),"SOURCE")
                .addSink("SINK",to,"PROCESSOR");

        //创建kafka stream
        KafkaStreams kafkaStreams = new KafkaStreams(builder,streamsConfig);

        kafkaStreams.start();
        System.out.println("kafka stream started!!!");







    }
}
